package com.ohos.connect.devicecloud.mqtt.core.config;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.mqttv5.client.MqttClient;
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttSubscription;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManager;
import javax.net.ssl.TrustManagerFactory;
import java.io.*;
import java.security.*;
import java.security.cert.CertificateException;
import java.util.ArrayList;
import java.util.List;

@Component
@Slf4j
public class MqttsClientInit implements CommandLineRunner{

    public static MqttClient mqttsClient;

    @Value("${mqtts.broker}")
    public String MQTTS_BROKER;

    @Value("${mqtts.clientId}")
    public String MQTTS_CLIENT_ID;

    @Value("${mqtts.connection-timeout}")
    public String MQTTS_CONNECTION_TIMEOUT;

    @Value("${mqtts.mqttProtocalTls}")
    public String MQTTS_PROTOCAL_TLS;

    @Value("${mqtts.username}")
    public String MQTTS_USERNAME;

    @Value("${mqtts.password}")
    public String MQTTS_PASSWORD;

    @Value("${mqtts.keyPath}")
    public String KEY_STORE_PATH;


    @Value("${mqtts.keyPassword}")
    public String KEY_STORE_PASSWORD;

    @Resource
    private RedisTemplate<String, Object> redisTemplate;

    /** 所有缓存Key的前缀 */
    private static final String MQTT_USER_KEY_PREFIX = "mqtt_user";

    public static boolean clientIsOk = false;

    public void initMqttsClient() {

        // 2.往redis中插入认证信息
        redisTemplate.boundHashOps(MQTT_USER_KEY_PREFIX + ":" + MQTTS_CLIENT_ID).put("username", MQTTS_USERNAME);
        redisTemplate.boundHashOps(MQTT_USER_KEY_PREFIX + ":" + MQTTS_CLIENT_ID).put("password", MQTTS_PASSWORD);
        redisTemplate.boundHashOps(MQTT_USER_KEY_PREFIX + ":" + MQTTS_CLIENT_ID).put("clientId", MQTTS_CLIENT_ID);

        // 3.创建MQTT Client
        log.info("Connecting to EMQX broker: {}", MQTTS_BROKER);
        int connCount = 0;
        while (connCount < 5) {
            try {
                if(!clientIsOk){
                    mqttsClient = new MqttClient(MQTTS_BROKER, MQTTS_CLIENT_ID, new MemoryPersistence());

                    SSLContext sslContext = createSSLContext();

                    // MQTT 连接选项
                    MqttConnectionOptions options = new MqttConnectionOptions();
                    options.setUserName(MQTTS_USERNAME);
                    options.setPassword(MQTTS_PASSWORD.getBytes());
                    options.setConnectionTimeout(30);       // 设置为0，禁用连接超时
                    options.setHttpsHostnameVerificationEnabled(false);
                    mqttsClient.setCallback(new MqttsClientCallback());

                    options.setSocketFactory(sslContext.getSocketFactory());

                    // 保留会话
                    options.setCleanStart(true);


                    // 建立连接
                    mqttsClient.connect(options);
                    log.info("EMQX 第 {} 次连接成功",connCount+1);

                    // 订阅topic
                    List<MqttSubscription> topics = new ArrayList<>();
                    MqttSubscription testMqttSubscription = new MqttSubscription("/test/dt");
                    topics.add(testMqttSubscription);

                    MqttSubscription[] mqttSubscriptions = topics.toArray(new MqttSubscription[topics.size()]);
                    mqttsClient.subscribe(mqttSubscriptions);
                    clientIsOk=true;

                    break;
                }
            } catch (MqttException me) {
                log.warn("EMQX mqtts 第 {} 次尝试连接失败 ", connCount+1);
                log.error("连接出错",me);
                connCount++;
                try {
                    long sleepTime = 10000;     // 每10秒尝试一次
                    Thread.sleep(sleepTime);
                }catch (InterruptedException e){
                    log.error(e.getMessage());
                }
            } catch (Exception e) {
                log.warn("EMQX连接创建失败 ", e);
            }
        }

        if(connCount > 4){
            log.error("mqtt 5 次 连接失败");
            System.exit(0);
        }

    }



    private SSLContext createSSLContext() throws NoSuchAlgorithmException, KeyStoreException,
            CertificateException, IOException, KeyManagementException {
        // 创建SSL上下文
        SSLContext sslContext = SSLContext.getInstance("TLS");

        // 加载密钥库（如果需要）
        KeyStore keyStore = KeyStore.getInstance("JKS");
        keyStore.load(new FileInputStream(KEY_STORE_PATH), KEY_STORE_PASSWORD.toCharArray());

        // 初始化信任管理器
        TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
        trustManagerFactory.init(keyStore);
        TrustManager[] trustManagers = trustManagerFactory.getTrustManagers();

        // 初始化SSL上下文
        sslContext.init(null, trustManagers, new SecureRandom());
        return sslContext;
    }

    @Override
    public void run(String... args) {
        initMqttsClient();
    }

    public MqttClient getMqttsClient(){
        if(mqttsClient == null){
            initMqttsClient();
        }
        return mqttsClient;
    }

}
